/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering,
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 *
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.wsinfra.wsr.impl;

import java.util.Arrays;
import java.util.Hashtable;
import java.util.Vector;

import cgl.narada.wsinfra.WsMessageFlow;
import cgl.narada.wsinfra.WsProcessor;
import cgl.narada.wsinfra.exception.DeploymentException;
import cgl.narada.wsinfra.exception.IncorrectExchangeException;
import cgl.narada.wsinfra.exception.MessageFlowException;
import cgl.narada.wsinfra.exception.ProcessingException;
import cgl.narada.wsinfra.exception.UnknownExchangeException;
import cgl.narada.wsinfra.exception.WsFaultException;
import cgl.narada.wsinfra.schemas.addressing.ActionDocument;
import cgl.narada.wsinfra.schemas.addressing.EndpointReferenceType;
import cgl.narada.wsinfra.schemas.addressing.FromDocument;
import cgl.narada.wsinfra.schemas.addressing.ReplyToDocument;
import cgl.narada.wsinfra.schemas.soap.EnvelopeDocument;
import cgl.narada.wsinfra.schemas.wsr.PollRequestDocument;
import cgl.narada.wsinfra.schemas.wsr.RefToMessageIdsType;
import cgl.narada.wsinfra.schemas.wsr.RequestDocument;
import cgl.narada.wsinfra.schemas.wsr.ResponseDocument;
import cgl.narada.wsinfra.schemas.wsr.SequenceNumRangeType;
import cgl.narada.wsinfra.util.SoapMessageAlteration;
import cgl.narada.wsinfra.wsa.AddressingHeaders;
import cgl.narada.wsinfra.wsa.ParseWsaHeaders;
import cgl.narada.wsinfra.wsa.WsaEnvelopeCreator;
import cgl.narada.wsinfra.wsa.WsaEprRules;
import cgl.narada.wsinfra.wsa.WsaProcessingFactory;
import cgl.narada.wsinfra.wsr.WsrAckOperations;
import cgl.narada.wsinfra.wsr.WsrActions;
import cgl.narada.wsinfra.wsr.WsrConstants;
import cgl.narada.wsinfra.wsr.WsrElementAddition;
import cgl.narada.wsinfra.wsr.WsrElementCreation;
import cgl.narada.wsinfra.wsr.WsrExchangeInfo;
import cgl.narada.wsinfra.wsr.WsrExchangeInfoCreator;
import cgl.narada.wsinfra.wsr.WsrNodeUtils;
import cgl.narada.wsinfra.wsr.WsrProcessingFactory;
import cgl.narada.wsinfra.wsr.WsrQNames;
import cgl.narada.wsinfra.wsr.WsrRequestCreator;
import cgl.narada.wsinfra.wsr.WsrResponseCreator;
import cgl.narada.wsinfra.wsr.WsrSink;
import cgl.narada.wsinfra.wsr.storage.WsrGroupInfoStorageOperations;
import cgl.narada.wsinfra.wsr.storage.WsrProtocolStorageOperations;
import cgl.narada.wsinfra.wsr.storage.WsrStorageException;
import cgl.narada.wsinfra.wsr.storage.WsrStorageService;
import cgl.narada.wsinfra.wsr.storage.impl.WsrGroupInfoImpl;
import cgl.narada.wsinfra.wsr.storage.impl.WsrGroupTerminationInfoImpl;
import cgl.narada.wsinfra.wsr.storage.impl.WsrStorageWidgetImpl;


/**
 * This class outlines the functionality of a WSR Sink.
 *
 * @author Damodar Yemme, Sima Patel
 * $Created on February 16, 2005, 1:53 PM $
 * $Version 1.0 $
 */
public class WsrSinkNode extends WsProcessor implements WsrSink{

    private WsMessageFlow wsMessageFlow;
    private WsrSinkNodeHelper sinkNodeHelper;
    private EndpointReferenceType sinkEpr;
    private WsaEnvelopeCreator wsaEnvelopeCreator;
    private WsaEprRules wsaEprRules;
    private WsrQNames wsrQNames;
    private WsrExchangeInfoCreator wsrExchangeInfoCreator;
    private WsrAckOperations wsrAckOperations;
    private WsrProtocolStorageOperations wsrProtocolOps;
    private WsrElementAddition wsrElementAddition;
    private WsrElementCreation wsrElementCreation;
    private WsrRequestCreator wsrRequestCreator;
    private WsrNodeUtils wsrNodeUtils;
    private WsrActions wsrActions;
    private WsrSinkGroupTerminationImpl wsrSinkGroupTerminationImpl;
    private WsrUtil wsrUtil;
    private WsrStorageService wsrStorageServiceSink;
    private WsrGroupInfoImpl wsrGroupInfoImpl;
    private WsrGroupTerminationInfoImpl wsrGroupTerminationInfoImpl;
    private WsrGroupInfoStorageOperations wsrGroupInfoStorageOperations;
    private WsrProtocolStorageOperations wsrProtocolStorageOperations;
    private WsrStorageWidgetImpl wsrStorageWidgetImpl;
    private ParseWsaHeaders parseWsaHeaders;
    private WsrFaultsImpl wsrFaultsImpl;
    private WsrResponseCreatorImpl wsrResponseCreatorImpl;
    private WsrResponseCreator wsrResponseCreator;
    private SoapMessageAlteration soapMessageAlteration;
    private String moduleName = "WsrSinkNode: ";

    /** Creates a new instance of WsrSinkNode */
    public WsrSinkNode(String configInfo)throws DeploymentException {
        try {
            wsrStorageServiceSink = WsrStorageService.getSinkStorageServiceInstance(configInfo);
	    wsrProtocolStorageOperations =
                            wsrStorageServiceSink.getWsrProtocolStorageOperations();
            wsrGroupInfoStorageOperations =
                           wsrStorageServiceSink.getWsrGroupInfoStorageOperations();

        } catch (WsrStorageException wsrStorageException) {
            String reason = moduleName + "Problems initializing storage services"
                    + ", please check the location of the config file [" +
                    configInfo + "] that was provided.\n" +
                    wsrStorageException.toString();
            throw new DeploymentException(reason);
        }

        sinkNodeHelper = WsrSinkNodeHelper.getInstance();
        sinkNodeHelper.initialize(wsrStorageServiceSink);
        wsrSinkGroupTerminationImpl = WsrSinkGroupTerminationImpl.getInstance();
        wsrSinkGroupTerminationImpl.initialize(wsrStorageServiceSink);
        wsrFaultsImpl = WsrFaultsImpl.getInstance();
        wsrQNames = WsrQNames.getInstance();
        wsrActions = WsrActions.getInstance();
        wsrElementCreation = WsrProcessingFactory.getWsrElementCreation();
        wsrElementAddition = WsrProcessingFactory.getWsrElementAddition();
        wsrExchangeInfoCreator = WsrProcessingFactory.getWsrExchangeInfoCreator();
        wsrResponseCreator = WsrProcessingFactory.getWsrResponseCreator();
        wsrAckOperations = WsrProcessingFactory.getWsrAckOperations();
        wsrNodeUtils = WsrProcessingFactory.getWsrNodeUtils();
        wsrUtil = WsrUtil.getInstance();
        soapMessageAlteration = SoapMessageAlteration.getInstance();
        wsaEnvelopeCreator = WsaProcessingFactory.getWsaEnvelopeCreator();
        wsaEprRules = WsaProcessingFactory.getWsaEprRules();
      

    }
    /** Sets the message flow which the processor should use */
    public void setMessageFlow(WsMessageFlow wsMessageFlow)
                throws DeploymentException {
        this.wsMessageFlow = wsMessageFlow;
        sinkNodeHelper.setMessageFlow(wsMessageFlow);
    }
    /** Gets the message flow which the processor should use. */
    public WsMessageFlow getMessageFlow() {
        return wsMessageFlow;
    }
    /** Set the endpoint reference associated with this node */
    public void setEndpointReference(EndpointReferenceType endpointReference) {
        sinkEpr = endpointReference;
        sinkNodeHelper.setEndpointReference(sinkEpr);
    }
    /** get the endpoint reference associated with this node */
    public EndpointReferenceType getEndpointReference(){
        return sinkEpr;
    }
    /** Process the exchange. The argument also indicates the direction in
    * which the exchange has actually traversed. */
    public boolean processExchange(EnvelopeDocument envelopeDocument,
                                int direction)
                throws UnknownExchangeException, IncorrectExchangeException,
                       ProcessingException,  MessageFlowException{

	WsrExchangeInfo wsrExchangeInfo =
                 wsrExchangeInfoCreator.createWsrExchangeInfo(envelopeDocument);
        AddressingHeaders addressingHeaders =
                                         wsrExchangeInfo.getAddressingHeaders();
        boolean isFault = isFault(addressingHeaders);
        if (isFault) {
          return true;
        }
	sinkNodeHelper.checkExchangeType(wsrExchangeInfo, direction);
        
        try {
            /** Only valid exchanges to be processed now remain */
            if (direction == WsMessageFlow.FROM_APPLICATION) {
                /** We will not be processing any messages from the application.
                Only Response Payload from the application will be directed 
                to the network
                */
            
                processMessageFromApplication(envelopeDocument,
                                                             addressingHeaders);
               
                
                return true;

            }
            /** From here on every thing else has been received over the network */
	   if (wsrExchangeInfo.isProcessRequest() && 
                                                 wsrExchangeInfo.hasRequest()) {
                processRequest(envelopeDocument, addressingHeaders);
               
                return false;
            }else{
                if(wsrExchangeInfo.isProcessPollRequest() && 
                                              wsrExchangeInfo.hasPollRequest()){
                	
                    processPollRequest(envelopeDocument, addressingHeaders);
                    
                    return false;
                }
            }
        } catch (WsFaultException wsFaultException) {
            manageFaultException(wsFaultException, direction);
        } catch (WsrStorageException wsrStorageException) {
        /** Need to manage storage exception here */
            System.out.println(moduleName + wsrStorageException.toString() );
        }
        return true;
    }
    /** Process a simple request received from the source */
    public void  processRequest(EnvelopeDocument envelopeDocument,
                                AddressingHeaders addressingHeaders)
                 throws WsFaultException,  MessageFlowException,
                        WsrStorageException{
        RequestDocument requestDocument =
                              wsrNodeUtils.getRequestDocument(envelopeDocument);
        if(!sinkNodeHelper.checkForProblemsInRequestDocument(requestDocument,
                                                            addressingHeaders)){
            return;
        }
       
	String groupId = requestDocument.getRequest().getMessageId().getGroupId();
        if(wsrGroupInfoStorageOperations.isGroupExists(groupId)){
         	
             processExistingGroup(groupId, envelopeDocument, addressingHeaders);
	     
             
         }else{
         	
             processNewGroup(groupId, envelopeDocument, addressingHeaders);
            
         }
	 wsrSinkGroupTerminationImpl.processGroupTermination(groupId);
	}
    /** Process New Group Request*/
    public void processNewGroup(String groupId,
                                EnvelopeDocument envelopeDocument,
                                AddressingHeaders addressingHeaders)
                throws WsFaultException, WsrStorageException,
                       MessageFlowException{
    
        EnvelopeDocument newEnvelopeDocument = null;
        boolean isNonSequenceReply = false;
        RequestDocument requestDocument =
                              wsrNodeUtils.getRequestDocument(envelopeDocument);
        wsrStorageWidgetImpl =
           sinkNodeHelper.createMessageInfoFromRequestDocument(requestDocument);
        if(wsrStorageWidgetImpl.getSequenceNumber() != 0){
            String fault = wsrFaultsImpl.getGroupAborted().toString();
            newEnvelopeDocument = createResponseEnvelope(requestDocument,
                                                         addressingHeaders,
                                                         groupId, fault, false,
                                      wsrStorageWidgetImpl.getSequenceNumber());
         
            enrouteToNetwork(newEnvelopeDocument);
            sendForCallback(newEnvelopeDocument, requestDocument, addressingHeaders);           
            return;
        }
	
        wsrGroupInfoImpl = sinkNodeHelper.createGroupInfoFromRequestDocument(
                                                             requestDocument,
                                                             addressingHeaders);
        wsrGroupTerminationInfoImpl = new WsrGroupTerminationInfoImpl(groupId);
        wsrGroupTerminationInfoImpl.setIsSource(false);

        if(requestDocument.getRequest().getMessageId().getSequenceNum().
                                                                 isSetLast() &&
           requestDocument.getRequest().getMessageId().getSequenceNum().
                                                                     getLast()){
        	
            isNonSequenceReply = true;
	               
        }
	
        sinkNodeHelper.updateGroupTerminationInfo(
                                   wsrGroupTerminationInfoImpl,
                                   wsrGroupInfoImpl.getGroupExpiryTime(),
                                   wsrGroupInfoImpl.getGroupMaxIdleDuration(),
                                   wsrStorageWidgetImpl.getMessageExpiryTime());

        if(wsrGroupTerminationInfoImpl.isGroupClosed()){
            String fault = wsrFaultsImpl.getGroupAborted().toString();
            newEnvelopeDocument = createResponseEnvelope(requestDocument, 
                                                         addressingHeaders,
                                                         groupId, fault, 
                                                         isNonSequenceReply, 0);
           
            wsrProtocolStorageOperations.storeGroupTerminationInfo(
                                                   wsrGroupTerminationInfoImpl);
            wsrGroupInfoStorageOperations.storeCreatedGroup(wsrGroupInfoImpl);
            enrouteToNetwork(newEnvelopeDocument);
            sendForCallback(newEnvelopeDocument, requestDocument, addressingHeaders);
            return;
        }
        enrouteToApplication(envelopeDocument);
        
        wsrGroupInfoImpl.setSequenceNumberLastDelivered(0);
        newEnvelopeDocument = createResponseEnvelope(
                                             requestDocument, addressingHeaders,
                                             groupId, "ACK", 
                                             isNonSequenceReply, 0);
        
        wsrStorageWidgetImpl.setEnvelope(newEnvelopeDocument);
        wsrStorageWidgetImpl.setMessageStatus(WsrConstants.MESSAGE_ACKNOWLEDGED);
        wsrProtocolStorageOperations.store(wsrStorageWidgetImpl);
        if(requestDocument.getRequest().getMessageId().getSequenceNum().
                                                                  isSetLast() &&
				requestDocument.getRequest().getMessageId().getSequenceNum().
                                                                     getLast()){
        	wsrGroupTerminationInfoImpl.setLastMessageOfGroup(true);
        	wsrGroupInfoImpl.setLastSequenceNumber(0);
        	
        }
        wsrProtocolStorageOperations.storeGroupTerminationInfo(
                                                   wsrGroupTerminationInfoImpl);
        wsrGroupInfoStorageOperations.storeCreatedGroup(wsrGroupInfoImpl);
        enrouteToNetwork(newEnvelopeDocument);
        sendForCallback(newEnvelopeDocument, requestDocument, addressingHeaders);
       
        return;
    }
    public void sendForCallback(EnvelopeDocument envelopeDocument, 
                                     RequestDocument requestDocument,
                                     AddressingHeaders addressingHeaders)
                throws WsFaultException, MessageFlowException{
       if(requestDocument.getRequest().getReplyPattern().
                 getValue().toString().equalsIgnoreCase
                  (WsrConstants.REPLY_PATTERN_CALLBACK)){
           String replyTo = requestDocument.getRequest().getReplyPattern()
           		.getReplyTo().getReferenceScheme();
           String to = wsaEprRules.sendTo(addressingHeaders).getAddress().getStringValue();
           if(!replyTo.equalsIgnoreCase(to)){
               EnvelopeDocument newEnvelopeDocument1 = 
                   sinkNodeHelper.getSoapEnvelope(addressingHeaders, to);
               wsrResponseCreator.addResponse(newEnvelopeDocument1,
                       wsrNodeUtils.getResponseDocument(envelopeDocument));
               enrouteToNetwork(newEnvelopeDocument1);
           }
       }

    }
    /**Process existing group request*/
    public void processExistingGroup(String groupId,
                                     EnvelopeDocument envelopeDocument,
                                     AddressingHeaders addressingHeaders)
                throws WsFaultException, WsrStorageException,
                       MessageFlowException{
	
        long sequenceNumber;
        RequestDocument requestDocument = null;
        EnvelopeDocument newEnvelopeDocument = null;

        requestDocument = wsrNodeUtils.getRequestDocument(envelopeDocument);
        wsrGroupInfoImpl = 
                wsrGroupInfoStorageOperations.getGroupInfoUsingGroupId(groupId);
	
        wsrGroupTerminationInfoImpl = 
                  wsrProtocolStorageOperations.getGroupTerminationInfo(groupId);
	
        wsrStorageWidgetImpl = sinkNodeHelper.
                          createMessageInfoFromRequestDocument(requestDocument);
	
        sequenceNumber = wsrStorageWidgetImpl.getSequenceNumber();
	
        sinkNodeHelper.updateGroupTerminationInfo(
                                     wsrGroupTerminationInfoImpl,
                                     wsrGroupInfoImpl.getGroupExpiryTime(),
                                     wsrGroupInfoImpl.getGroupMaxIdleDuration(),
                                     null);
	if(wsrGroupTerminationInfoImpl.isGroupClosed()){
            String fault = wsrFaultsImpl.getGroupAborted().toString();
            newEnvelopeDocument = createResponseEnvelope(requestDocument, 
                                                         addressingHeaders,
                                                         groupId, fault, false,
                                                         sequenceNumber);
            
            wsrProtocolStorageOperations.storeGroupTerminationInfo(
                                                   wsrGroupTerminationInfoImpl);
            enrouteToNetwork(newEnvelopeDocument);
            sendForCallback(newEnvelopeDocument, requestDocument, addressingHeaders);
            return;
        }
	
        if(wsrProtocolStorageOperations.hasStoredElement(groupId,sequenceNumber)){
            if(wsrProtocolStorageOperations.
               getMessageStatus(groupId,sequenceNumber)!=
                                            WsrConstants.MESSAGE_NOT_DELIVERED){
                WsrStorageWidgetImpl wsrStorageWidgetImpl1 =
                wsrProtocolStorageOperations.getStoredElement(
                                                        groupId,sequenceNumber);
                enrouteToNetwork(wsrStorageWidgetImpl1.getEnvelope());
                sendForCallback(wsrStorageWidgetImpl1.getEnvelope(), 
                        requestDocument, addressingHeaders);
                return;
            }
         }

        
        long previousSequenceNumber = wsrGroupInfoStorageOperations.
                                        getSequenceNumberLastDelivered(groupId);
        if(sequenceNumber == previousSequenceNumber+1){
        //first check for messageExpirytime is message is expired do not 
        //deliever it send the group aborted fault.
        if(wsrUtil.isExpiryTimePassed(
                                 wsrStorageWidgetImpl.getMessageExpiryTime())){
            wsrGroupTerminationInfoImpl.setMessageExpired(true);
            String fault = wsrFaultsImpl.getGroupAborted().toString();
            newEnvelopeDocument = createResponseEnvelope(requestDocument, 
                                                         addressingHeaders,
                                                         groupId, fault, false,
                                                                                                                     sequenceNumber);
            
            wsrProtocolStorageOperations.storeGroupTerminationInfo(
                                                   wsrGroupTerminationInfoImpl);
            enrouteToNetwork(newEnvelopeDocument);
            sendForCallback(newEnvelopeDocument, requestDocument, addressingHeaders);
            return;

        }
        enrouteToApplication(envelopeDocument);
        wsrGroupInfoImpl.setSequenceNumberLastDelivered(sequenceNumber);
        newEnvelopeDocument = createResponseEnvelope(requestDocument, 
                                                     addressingHeaders,
                                                     groupId, "ACK", false,
                                                     sequenceNumber);
        wsrStorageWidgetImpl.setEnvelope(newEnvelopeDocument);
        wsrStorageWidgetImpl.setMessageStatus(WsrConstants.MESSAGE_ACKNOWLEDGED);
        wsrProtocolStorageOperations.store(wsrStorageWidgetImpl);
        enrouteToNetwork(newEnvelopeDocument); 
        sendForCallback(newEnvelopeDocument, requestDocument, addressingHeaders);
        if(sequenceNumber ==  Long.MAX_VALUE){
        	
            wsrGroupTerminationInfoImpl.setMaxSequenceNumber(true);
            wsrGroupInfoStorageOperations.storeCreatedGroup(wsrGroupInfoImpl);
            wsrProtocolStorageOperations.storeGroupTerminationInfo(
                                                   wsrGroupTerminationInfoImpl);
            return;
        }
        boolean lastmessage = 
        	requestDocument.getRequest().getMessageId().getSequenceNum().getLast();
       
        if(requestDocument.getRequest().getMessageId().getSequenceNum().
                isSetLast()&&
           requestDocument.getRequest().getMessageId().getSequenceNum().
                getLast()==true){
            wsrGroupTerminationInfoImpl.setLastMessageOfGroup(true);
            wsrGroupInfoImpl.setLastSequenceNumber(sequenceNumber);
            wsrGroupInfoStorageOperations.storeCreatedGroup(wsrGroupInfoImpl);
            wsrProtocolStorageOperations.storeGroupTerminationInfo(
                                                   wsrGroupTerminationInfoImpl);
            return;
        }
        wsrGroupInfoStorageOperations.storeCreatedGroup(wsrGroupInfoImpl);
        wsrProtocolStorageOperations.storeGroupTerminationInfo(
                                               wsrGroupTerminationInfoImpl);
        processUndeliveredSequenceNumbers(groupId, sequenceNumber,
                                          addressingHeaders,requestDocument);
        wsrProtocolStorageOperations.storeGroupTerminationInfo(
                                                   wsrGroupTerminationInfoImpl);
        wsrGroupInfoStorageOperations.storeCreatedGroup(wsrGroupInfoImpl);
        return;

        }else{
            if(sequenceNumber > previousSequenceNumber+1){
                if(wsrGroupInfoImpl.hasLastSequenceNumber()&&
                    sequenceNumber >= wsrGroupInfoImpl.getLastSequenceNumber()){
                    String fault = wsrFaultsImpl.getGroupAborted().toString();
                    newEnvelopeDocument = createResponseEnvelope(requestDocument,
                                                          addressingHeaders,
                                                          groupId, fault, false,
                                                                                                                     sequenceNumber);
                    enrouteToNetwork(newEnvelopeDocument);
                    sendForCallback(newEnvelopeDocument, 
                            requestDocument, addressingHeaders);
                    return;
                }else{

                    wsrStorageWidgetImpl.setMessageStatus(
                                            WsrConstants.MESSAGE_NOT_DELIVERED);
                    wsrProtocolStorageOperations.store(wsrStorageWidgetImpl);
                    
                    String fault = 
                         wsrFaultsImpl.getMessageProcessingFailure().toString();
                    newEnvelopeDocument = createResponseEnvelope(
                                                          requestDocument,
                                                          addressingHeaders,
                                                          groupId, fault, false,
                                                                                                                                 sequenceNumber);
                    enrouteToNetwork(newEnvelopeDocument);
                    sendForCallback(newEnvelopeDocument, 
                            requestDocument, addressingHeaders);
                    return;
                }
            }else{
               wsrStorageWidgetImpl=
                      wsrProtocolStorageOperations.getStoredElement(groupId,
                                                                sequenceNumber);
               enrouteToNetwork(wsrStorageWidgetImpl.getEnvelope());
               sendForCallback(wsrStorageWidgetImpl.getEnvelope(), 
                       requestDocument, addressingHeaders);
               return;
            }

        }
    }

    /** process Undelivered Sequence Numbers */
    public void processUndeliveredSequenceNumbers(String groupId, 
                         long sequenceNumber,
                         AddressingHeaders addressingHeaders,
                         RequestDocument requestDocument)
                throws WsrStorageException, MessageFlowException,
                       WsFaultException{
    	
        EnvelopeDocument newEnvelopeDocument = null;
        EnvelopeDocument newEnvelopeDocument1 = null;
        long [] sequenceNumbers =
                     wsrProtocolStorageOperations.getUnDeliveredSequenceNumbers(
                                                       groupId, sequenceNumber);
        
        wsrGroupInfoImpl = 
        	wsrGroupInfoStorageOperations.getGroupInfoUsingGroupId(groupId);
        wsrGroupTerminationInfoImpl = 
        	wsrProtocolStorageOperations.getGroupTerminationInfo(groupId);
        if(sequenceNumbers == null){
        	 	return;
        }        	
        Arrays.sort(sequenceNumbers);
        
        if(sequenceNumbers.length != 0 && sequenceNumbers != null){
            for(int index = 0; index < sequenceNumbers.length; index++){
                long storedSequenceNumber = 
                              wsrGroupInfoImpl.getSequenceNumberLastDelivered();
                
                if(storedSequenceNumber+1 == sequenceNumbers[index]){
                    wsrStorageWidgetImpl =
                                wsrProtocolStorageOperations.getStoredElement(
                                               groupId, sequenceNumbers[index]);
                    sinkNodeHelper.updateGroupTerminationInfo(
                                   wsrGroupTerminationInfoImpl,
                                   wsrGroupInfoImpl.getGroupExpiryTime(),
                                   wsrGroupInfoImpl.getGroupMaxIdleDuration(),
                                   wsrStorageWidgetImpl.getMessageExpiryTime());
                    if(wsrGroupTerminationInfoImpl.isGroupClosed()){
                        String fault = 
                                     wsrFaultsImpl.getGroupAborted().toString();
                        newEnvelopeDocument = createResponseEnvelope(
                                                        requestDocument,
                                                        addressingHeaders,
                                                        groupId, fault, false,
                                                        sequenceNumbers[index]);
                         enrouteToNetwork(newEnvelopeDocument);
                         sendForCallback(newEnvelopeDocument, 
                                 requestDocument, addressingHeaders);
                         wsrStorageWidgetImpl.setMessageFault(
                                               wsrFaultsImpl.getGroupAborted());
                         wsrProtocolStorageOperations.store(
                                                          wsrStorageWidgetImpl);
                         return;
                    }
                    enrouteToApplication(wsrStorageWidgetImpl.getEnvelope());
                     wsrGroupInfoImpl.setSequenceNumberLastDelivered(
                                                        sequenceNumbers[index]);
                    newEnvelopeDocument1 = wsrStorageWidgetImpl.getEnvelope();
                    RequestDocument newRequestDocument=
                          wsrNodeUtils.getRequestDocument(newEnvelopeDocument1);
                    newEnvelopeDocument = createResponseEnvelope(
                                             newRequestDocument, 
                                             addressingHeaders,
                                             groupId, "ACK", false,
                                             sequenceNumber);
                    enrouteToNetwork(newEnvelopeDocument);
                    sendForCallback(newEnvelopeDocument, 
                            requestDocument, addressingHeaders);
                    wsrStorageWidgetImpl.setEnvelope(newEnvelopeDocument);
                    wsrStorageWidgetImpl.setMessageStatus(
                                             WsrConstants.MESSAGE_ACKNOWLEDGED);
                    wsrProtocolStorageOperations.store(wsrStorageWidgetImpl);
                    if(sequenceNumbers[index] == Long.MAX_VALUE){
                        wsrGroupTerminationInfoImpl.setMaxSequenceNumber(true);
                    }
                    if(newRequestDocument.getRequest().getMessageId().
                                                 getSequenceNum().isSetLast()&&
                        requestDocument.getRequest().getMessageId().
                                                    getSequenceNum().getLast()){
                        wsrGroupTerminationInfoImpl.setLastMessageOfGroup(true);
                        wsrGroupInfoImpl.setLastSequenceNumber(
                                                        sequenceNumbers[index]);
                    }
                }
            }
        }
    }
    public EnvelopeDocument createResponseEnvelope(RequestDocument requestDocument,
                                            AddressingHeaders addressingHeaders,
                                            String groupId, String fault,
                                            boolean isNonSequenceReply,
                                            long sequenceNumber)
                            throws MessageFlowException{
        ResponseDocument responseDocument = null;
        if(isNonSequenceReply){
            if(fault.equalsIgnoreCase("ACK")){
            	responseDocument=wsrElementCreation.newResponse(responseDocument,
            	                                                groupId,"ACK");
            }else{
            	responseDocument=wsrElementCreation.newResponse(responseDocument,
            	                                                groupId,fault);
            }
        }else{
            if(fault.equalsIgnoreCase("ACK")){
            	responseDocument=wsrElementCreation.newResponse(responseDocument, 
            	                            groupId,
                                          sequenceNumber, sequenceNumber,"ACK");
            }else{
            	responseDocument=wsrElementCreation.newResponse(responseDocument, 
            	                            groupId,
                                          sequenceNumber, sequenceNumber,fault);
            }
        }
        EnvelopeDocument newEnvelopeDocument =
                                sinkNodeHelper.createResponse(addressingHeaders,
                                                              requestDocument,
                                                              responseDocument);

        return newEnvelopeDocument;

    }
    /** Process exchange received from application, and send it across
    * reliability */
    public void processMessageFromApplication(EnvelopeDocument envelopeDocument,
                                            AddressingHeaders addressingHeaders)
                throws WsFaultException,  MessageFlowException,
                       WsrStorageException{
        //simply enroute to network
        enrouteToNetwork(envelopeDocument);
    }

    /** Process a PollRequest received from the source for Acknowledgements*/
    public void processPollRequest(EnvelopeDocument envelopeDocument,
                                   AddressingHeaders addressingHeaders)
                throws WsFaultException,  MessageFlowException,
                       WsrStorageException{
        PollRequestDocument pollRequestDocument =
                          wsrNodeUtils.getPollRequestDocument(envelopeDocument);
        sinkNodeHelper.checkForProblemsInPollRequestDocument(
                                        pollRequestDocument, addressingHeaders);
        ResponseDocument responseDocument = wsrElementCreation.newResponse();
        RefToMessageIdsType[] refToMessageIdsType =
                 pollRequestDocument.getPollRequest().getRefToMessageIdsArray();
        int noOfElementsAdded = 0;
        for(int count = 0;count < refToMessageIdsType.length; count++){
            String groupId = refToMessageIdsType[count].getGroupId();
           
            if(refToMessageIdsType[count].sizeOfSequenceNumRangeArray()!=0){
                SequenceNumRangeType[] sequenceNumRangeType =
                          refToMessageIdsType[count].getSequenceNumRangeArray();
                Vector seqNumVector = new Vector();
                for(int index = 0; index < sequenceNumRangeType.length;index++){
                    long from=sequenceNumRangeType[index].getFrom().longValue();
                    long to = sequenceNumRangeType[index].getTo().longValue();
                    while(from <= to){
                        seqNumVector.addElement(new Long(from));
                        from++;
                    }
                }
                long[] seqNums = createLongArrayFromVector(seqNumVector);
                if(addStatusToResponse(responseDocument, groupId,seqNums)){
                	
                    noOfElementsAdded++;
                }
            }else{
            //retrieve the status of all the messages of that group in question
                if(addStatusToResponse(responseDocument, groupId, null)){
                    noOfElementsAdded++;
                }
            }
        }
        if(noOfElementsAdded != 0){
        	
            String sourceAddress = pollRequestDocument.getPollRequest().
                                              getReplyTo().getReferenceScheme();
            EndpointReferenceType sourceEpr =
                                    EndpointReferenceType.Factory.newInstance();
            sourceEpr.addNewAddress().setStringValue(sourceAddress);
            FromDocument from = FromDocument.Factory.newInstance();
            from.setFrom(sinkEpr);
         
            ActionDocument action = ActionDocument.Factory.newInstance();
            action.addNewAction().setStringValue(wsrActions.getProcessResponse());
         
            ReplyToDocument replyTo = ReplyToDocument.Factory.newInstance();
            replyTo.setReplyTo(sinkEpr);
            EnvelopeDocument newEnvelopeDocument =
            	wsaEnvelopeCreator.createSoapEnvelope(sourceEpr, from, action,
                                                 replyTo, null, null);
            soapMessageAlteration.addToSoapHeaderAsLastChild(
                                         newEnvelopeDocument, responseDocument);
            
            enrouteToNetwork(newEnvelopeDocument);
        }
    }
    public boolean addStatusToResponse(ResponseDocument responseDocument,
                                       String groupId, long[] seqNums)
                   throws WsrStorageException{
    	
        Hashtable seqNumsAndStatus = new Hashtable();
        
        boolean isNonSequenceReply = false;
        if(seqNums == null){
            if(wsrProtocolStorageOperations.hasStoredElements(groupId)){
                seqNums = wsrProtocolStorageOperations.getSequenceNumbers(
                                                                       groupId);
            }else{
                return false;
            }
        }
        if(seqNums.length == 1){
        	
        	WsrGroupInfoImpl wsrGroupInfoImpl1 = 
            	wsrGroupInfoStorageOperations.getGroupInfoUsingGroupId(groupId);
        	if(wsrGroupInfoImpl1 == null){
        		return false;
        	}
            if(wsrGroupInfoImpl1.hasLastSequenceNumber()== true &&
            	wsrGroupInfoImpl1.getLastSequenceNumber()== 0 && seqNums[0] == 0){
                isNonSequenceReply = true;
                
            }
        }
        if(isNonSequenceReply){
            byte messageStatus =
                          wsrProtocolStorageOperations.getMessageStatus(groupId,
                                                                    seqNums[0]);
            if(messageStatus == WsrConstants.MESSAGE_ACKNOWLEDGED){
                wsrElementCreation.newResponse(responseDocument, groupId,"ACK");
                return true;
            }else if(messageStatus == WsrConstants.MESSAGE_FAULTED){
                String fault =  wsrProtocolStorageOperations.getMessageFault(
                                                           groupId, seqNums[0]);
                wsrElementCreation.newResponse(responseDocument,groupId,fault);
                return true;
            }else{
                return false;
            }
        }
        //it is sequencereplies
        for(int index = 0;index < seqNums.length; index++){
            byte messageStatus =
                          wsrProtocolStorageOperations.getMessageStatus(groupId,
                                                                seqNums[index]);
            if(messageStatus == WsrConstants.MESSAGE_ACKNOWLEDGED){
                seqNumsAndStatus.put(new Long(seqNums[index]), "ACK");
            }
            if(messageStatus == WsrConstants.MESSAGE_FAULTED){
                String fault =  wsrProtocolStorageOperations.getMessageFault(
                                                       groupId, seqNums[index]);
                seqNumsAndStatus.put(new Long(seqNums[index]), fault);
            }
        }
        Hashtable rangesAndFaults = wsrAckOperations.createRanges(
                                                              seqNumsAndStatus);
        if(rangesAndFaults == null){
            return false;
        }else{
            wsrElementCreation.newResponse(responseDocument, groupId,
                                           rangesAndFaults);
            return true;
        }
    }

    /** create Long Number array from Vector */
    public long[] createLongArrayFromVector(Vector vector){
        long[] numbers = new long[vector.size()];
        for(int index=0;index<vector.size();index++){
            numbers[index]   = ((Long)(vector.elementAt(index))).longValue();
        }
        return numbers;
    }

}
